package flow

import (
	"encoding/json"
	"github.com/s8sg/mini-flow/sdk"
	"os"
	"sync"
)

type FlowExecutor struct {
	Flow *Workflow
}

//type Message struct {
//	data map[string][]byte `json:"data"`
//}

type Bolt struct {
	value []byte `json:"value"`
	name  string `json:"name"`
}

func (fexec *FlowExecutor) ExecuteFlow(request []byte) ([]byte, error) {
	var result []byte
	var err error
	var resultMap map[string][]byte

	options := make(map[string]interface{})
	options["gateway"] = os.Getenv("gateway")
	options["request-id"] = os.Getenv("request-id")

	workflow := fexec.Flow

	var waitAll sync.WaitGroup
	waitAll.Add(workflow.GetNodeLeft())

	for workflow.GetNodeLeft() != 0 {
		results := make(chan *Bolt)
		startNodes := workflow.GetStartNodes()
		startNodeIds := make([]string, 0)
		var waitCurrentLevel sync.WaitGroup
		waitCurrentLevel.Add(startNodes.Len())
		for item := startNodes.Front();nil != item ;item = item.Next() {
			node := item.Value.(*sdk.Node)
			startNodeIds = append(startNodeIds, node.Id)
			go func(execNode *sdk.Node) {
				input, output := node.Offer()
				if len(input)>0 {
					inputMap := make(map[string][]byte)
					for _, v := range input {
						if resultMap[v] == nil {
							continue
						}
						inputMap[v] = resultMap[v]
					}
					if len(inputMap)>0 {
						result, _ = json.Marshal(inputMap)
					}
				}else{
					result = nil
				}
				for _, operation := range node.Operations() {
					if result == nil {
						result, err = operation.Execute(request, options)
					} else {
						result, err = operation.Execute(result, options)
					}
					if result != nil {
						outputRespMap := make(map[string][]byte)
						json.Unmarshal(result, &outputRespMap)
						counter := 0
						for _, v := range output {
							if outputRespMap[v] == nil {
								continue
							}
							results <- &Bolt{name: v, value: outputRespMap[v]}
							counter++
						}
						if counter == 0 {
							results <- &Bolt{name: node.Id, value: result}
						}
					}
				}
				waitCurrentLevel.Done()
				waitAll.Done()
			}(node)
		}
		go func() {
			waitCurrentLevel.Wait()
			close(results)
			workflow.RemoveExec(startNodeIds)
		}()
		resultMap = display(results)
	}
	waitAll.Wait()
	return result, err
}

func display(results chan *Bolt) map[string][]byte{
	data := make(map[string][]byte)
	for result := range results {
		data[result.name] = result.value
	}
	//resp, err := json.Marshal(data)
	//if err != nil {
	//	panic(err)
	//}
	return data
}
